package com.larry.spark.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 * 编号	字段名称	字段类型	字段含义
1	date	String	用户点击行为的日期
2	user_id	Long	用户的ID
3	session_id	String	Session的ID
4	page_id	Long	某个页面的ID
5	action_time	String	动作的时间点
6	search_keyword	String	用户搜索的关键词
7	click_category_id	Long	某一个商品品类的ID
8	click_product_id	Long	某一个商品的ID
9	order_category_ids	String	一次订单中所有品类的ID集合
10	order_product_ids	String	一次订单中所有商品的ID集合
11	pay_category_ids	String	一次支付中所有品类的ID集合
12	pay_product_ids	String	一次支付中所有商品的ID集合
13	city_id	Long	城市 id

 */
object PageFlow {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    val sc = new SparkContext(conf)

    val fileData = sc.textFile("data/user_visit_action.txt")

    //映射为样例类
    val actData = fileData.map(
      data => {
        val datas = data.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )

    actData.cache()

    //所有点击
    val clickAll: Map[Long, Int] = actData.map(
      d => {
        (d.page_id, 1)
      }
    ).reduceByKey(_ + _).collect().toMap



    val groupRDD: RDD[(String, Iterable[UserVisitAction])] = actData.groupBy(_.session_id)

    //组内排序
    val mapRDD: RDD[(String, List[(Long, Long)])] = groupRDD.mapValues(
      it => {
        val actions = it.toList.sortBy(_.action_time)
        val pageIds = actions.map(_.page_id)
        //滑动窗口
        //        val iterator = pageIds.sliding(2)
        //        while (iterator.hasNext) {
        //          val longs = iterator.next()
        //          (longs.head,longs.last)
        //        }
        //拉链
        val flowIds: List[(Long, Long)] = pageIds.zip(pageIds.tail)
        flowIds
      }
    )

    val mapRDD2 = mapRDD.map(_._2)
    val flatMap = mapRDD2.flatMap(list => list)

    val reduceRDD: RDD[((Long, Long), Int)] = flatMap.map((_, 1)).reduceByKey(_ + _)

    reduceRDD.foreach{
      case ((i1,i2),count) => {
        println( i1 + "-->" + i2 + " : " + (count.toDouble / clickAll.getOrElse(i1,1)))
      }
    }

//    reduceRDD.foreach(println)
    sc.stop()
  }

}
